package org.dyz.banana.controllers;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;
import org.dyz.banana.listeners.DemoListener;
import org.dyz.banana.vo.JsonResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.Set;
import java.util.concurrent.ExecutionException;

@RestController
public class TestInfoController {

    private static final Logger log= LoggerFactory.getLogger(TestInfoController.class);
    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;
    @Autowired
    private AdminClient adminClient;


    @GetMapping("/kafka/topic/listAll")
    public JsonResponse getTopicList(){
        StringBuilder retStr = new StringBuilder();
        ListTopicsResult result = adminClient.listTopics();
        KafkaFuture<Set<String>> names = result.names();
        try {
            names.get().forEach((k)->{
                retStr.append(k).append("\n");
            });
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
//        adminClient.close();
        return JsonResponse.success(retStr.toString());
    }

    @PostMapping("/kafka/topic/add/{topicName}")
    public JsonResponse addTopic(@PathVariable("topicName")String topicName){
        NewTopic[] nts = new NewTopic[]{new NewTopic(topicName,8,(short)1)};
        adminClient.createTopics(Arrays.asList(nts));
//        adminClient.close();
        return JsonResponse.success();
    }

    @PostMapping("/kafka/topic/delete/{topicName}")
    public JsonResponse deleteTopic(@PathVariable("topicName")String topicName){
        String[] nts = new String[]{topicName};
        adminClient.deleteTopics(Arrays.asList(nts));
//        adminClient.close();
        return JsonResponse.success();
    }

}
